package cloud.xlh.digital_power_order.disruptor;

import cloud.xlh.digital_power_order.infrastructure.order.entity.OrderEntity;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Created with IntelliJ IDEA.
 *
 * @ project name : xlh_osms
 * @ Author: XuLeHuang
 * @ Date: 2023/8/23 22:49
 * @ Description:
 */
@Configuration
public class RingBufferManager {

    private static final int ORDERHANDLER_ARRAY_LEN = 10;
    private static final String ORDERHANDLER_ID_PREFIX = "OrderHandler";

    @Resource(name = "IOThreadPoolExecutor")
    private ThreadPoolExecutor threadPoolExecutor;

    @Bean("orderEntityMessageModelRingBuffer")
    public RingBuffer<OrderEntity> orderEntityMessageModelRingBuffer() {
        //定义用于事件处理的线程池， Disruptor通过java.util.concurrent.ExecutorSerivce提供的线程来触发consumer的事件处理
        /*ExecutorService executor = Executors.newFixedThreadPool(2);*/

        //指定事件工厂
        EventFactory<OrderEntity> factory = new OrderFactory();

        //指定ringbuffer字节大小，必须为2的N次方（能将求模运算转为位运算提高效率），否则将影响效率
        int bufferSize = 1024 * 256;

        //单线程模式，获取额外的性能
        Disruptor<OrderEntity> disruptor = new Disruptor<>(factory, bufferSize, threadPoolExecutor,
                ProducerType.SINGLE, new BlockingWaitStrategy());

        //设置事件业务处理器---消费者

        //disruptor.handleEventsWith(new OrderHandler("1") , new OrderHandler("2"));
        /**
         * 多消费者不重复模式
         **/
        OrderHandler[] orderHandlers = new OrderHandler[ORDERHANDLER_ARRAY_LEN];
        for (int i = 0 ; i < orderHandlers.length ; i ++) {
            orderHandlers[i] = new OrderHandler(ORDERHANDLER_ID_PREFIX.concat(String.valueOf(i)));
        }
        disruptor.handleEventsWithWorkerPool(orderHandlers);

        // 启动disruptor线程
        disruptor.start();

        //获取ringbuffer环，用于接取生产者生产的事件
        RingBuffer<OrderEntity> ringBuffer = disruptor.getRingBuffer();

        return ringBuffer;
    }
}
